package top.caozj.task;

import lombok.extern.log4j.Log4j2;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import top.caozj.App;
import top.caozj.conf.AppConfig;
import top.caozj.conf.DataQuery;
import top.caozj.conf.Datasource;
import top.caozj.pojo.RawData;
import top.caozj.util.DataSourceType;

import java.io.UnsupportedEncodingException;
import java.sql.*;
import java.text.ParseException;
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 数据库查询任务类
 */
@Log4j2
public class QueryDataTask {
    //创建传感器数据管理对象,筛选出离线传感器
    public static SensorDataManager sensorDataManager =
            new SensorDataManager(
            AppConfig.sensorCodeMap.keySet(),
            AppConfig.noDataPolicyObject,
            AppConfig.fixedValue);

    // mqtt数据发送对象
    public MqttUpload mqttUpload;

    {
        try {
            mqttUpload = new MqttUpload();
        } catch (UnsupportedEncodingException e) {
            log.error("mqtt服务器配置错误");
            System.exit(-1);
        }
    }

    /**
     * 线程池，用于执行sql查询和mqtt数据发送的耗时任务
     */
    private static final ExecutorService executorService = Executors.newFixedThreadPool(10);

    /**
     *
     */
    public void run() throws InterruptedException {
        //统一查询起始时间
        Instant lastUpdateTime=Instant.now();
        ZonedDateTime zonedDateTime1 = lastUpdateTime.atZone(ZoneId.systemDefault());

        //查询开始时间为AppConfig.SQL_INTERVAL分之前，可以配置
//       这里之所以需要设置秒为0，是因为计算得到的整10分钟有可能会为01秒，比如，15:00:00，有可能系统会计算成15:00:01，从而导致查询结果为空。
        lastUpdateTime = zonedDateTime1.minusMinutes(AppConfig.SQL_INTERVAL).withSecond(0).toInstant();


        ZoneId zoneId = ZoneId.systemDefault();
        ZonedDateTime zonedDateTime = lastUpdateTime.atZone(zoneId);
        DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        String lastUpdateTimeStr = zonedDateTime.format(dateTimeFormatter);

        log.debug("current time:"+Instant.now()+ "    query start time:"+lastUpdateTimeStr);

        sensorDataManager.reset();


        //遍历数据源，执行查询
        AppConfig.datasource.forEach(datasource -> {
//            获取数据库类型，目前只支持MYSQL
            DataSourceType type = DataSourceType.valueOf(datasource.getType().toUpperCase());

            switch (type) {
                case MYSQL:
                    try {
                        //执行每个数据源执行对应的查询语句
                        runQueries(datasource,datasource.getDataQueries(), lastUpdateTimeStr, type);
                    } catch (Exception e) {
                        log.error("MySQL JDBC error");
//                        e.printStackTrace();
                    }

                    break;
                case REDIS:
                    log.info("use redis drive");
                    break;
                default:
                    log.error("unknown data source type");
                    break;

            }


        });


        Thread.sleep(3000);
        Set<String> offSetSensors = sensorDataManager.getOffSetSensors();
        log.info("开始处理离线传感器，共{}个",offSetSensors.size());
        log.info("当前策略：{}，默认值{}", AppConfig.noDataPolicyObject, AppConfig.fixedValue);

        String finalLastUpdateTime = lastUpdateTimeStr;
        offSetSensors.forEach(sensorId -> {
           mqttUpload.processOffSetData(sensorId, finalLastUpdateTime,sensorDataManager.getFixedValue());
        });


    }



//    1.查询数据，每条单独发送,多线程查询
    public void runQueries(Datasource datasource, List<DataQuery> dataQueries, String lastUpdateTime, DataSourceType type){
        switch (type){
            case MYSQL:
                dataQueries.forEach(dataQuery -> {
                    String newSQl = dataQuery.getSql().replace("${startTime}", "'" +lastUpdateTime+ "'");


                    //每条SQL启动一个线程去执行
                    executorService.execute(()->{
                        log.info(newSQl);
                        try {
                            Connection conn = DriverManager.getConnection(datasource.getUrl(), datasource.getUsername(), datasource.getPassword());
                            log.debug("connect to datasource: {}", datasource.getUrl());
                            PreparedStatement statement = conn.prepareStatement(newSQl);
                            ResultSet resultSet = statement.executeQuery();
                            int rowCount = 0;

                            while (resultSet.next()) {
                                rowCount++;
                                String id = resultSet.getString("id");
                                String value = resultSet.getString("value");
                                String tTime = resultSet.getString("t_time");
                                log.debug("id: {}, value: {}, t_time: {}", id, value, tTime);
                                //原始数据处理并通过mqtt协议发送

                                //不调试，则发送数据
                                if (!AppConfig.DEBUG){
                                    mqttUpload.processRawData(new RawData(id, value, tTime));
                                }

                                //标记传感器已处理
                                sensorDataManager.markSensorAsProcessed(id);

                            }

                            log.info("return record size: {} ", rowCount);
                            } catch (Exception e) {
                            log.error(e.getMessage());
                        }

                    });

                });
                break;
            case REDIS:
                log.info("use redis drive");
                break;
            default:
                log.error("unknown data source type");
                break;
        }

    }
}
